package org.truenewx.tnxjee.repo.redis.concurrent;

import java.util.concurrent.ExecutorService;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.truenewx.tnxjee.core.util.ExceptionUtil;
import org.truenewx.tnxjee.core.util.LogUtil;
import org.truenewx.tnxjee.core.util.concurrent.BatchTask;
import org.truenewx.tnxjee.repo.redis.support.RedisLockKeyPrefixSupplier;
import org.truenewx.tnxjee.repo.redis.support.RedisLockTemplate;

/**
 * 基于Redis同步锁的任务执行器
 *
 * @param <S> 业务范围类型，Scope的首字母
 */
public abstract class RedisLockTaskExecutor<S> {

    @Autowired
    private ExecutorService executorService;
    private final RedisLockTemplate lockTemplate;

    public RedisLockTaskExecutor(ApplicationContext applicationContext,
            RedisLockKeyPrefixSupplier redisPrefixSupplier) {
        this.lockTemplate = new RedisLockTemplate(applicationContext, redisPrefixSupplier);
        setDefaultLockExpiredMilliseconds(24 * 60 * 60 * 1000); // 同步锁的默认过期时间为24小时
    }

    public final void setDefaultLockExpiredMilliseconds(long defaultLockExpiredMilliseconds) {
        this.lockTemplate.setDefaultExpiredMilliseconds(defaultLockExpiredMilliseconds);
    }

    public final boolean isLocked(S key) {
        String lockKey = getLockKey(key);
        return this.lockTemplate.isLocked(lockKey);
    }

    /**
     * 同步锁校验通过后提交指定任务至线程池等待执行
     *
     * @param key                 同步锁键
     * @param expiredMilliseconds 同步锁超时毫秒数
     * @param throwOnRunning      当有正在运行的相同任务时是否抛出异常
     * @param task                任务
     */
    public void submit(S key, long expiredMilliseconds, boolean throwOnRunning, Runnable task) {
        String lockKey = getLockKey(key);
        if (!this.lockTemplate.tryLock(lockKey, expiredMilliseconds)) {
            if (throwOnRunning) {
                Exception exception = getLockedException(key);
                if (exception != null) {
                    throw ExceptionUtil.toRuntimeException(exception);
                }
            } else {
                LogUtil.info(getClass(), "{}({}) is running, break", key, lockKey);
            }
            return;
        }
        if (task instanceof BatchTask) {
            BatchTask batchTask = (BatchTask) task;
            // 任务每执行一步，都判断同步锁是否仍然在锁，以判断是否可以继续执行
            batchTask.setContinuePredicate((index, total) -> {
                boolean locked = this.lockTemplate.isLocked(lockKey);
                if (!locked) {
                    LogUtil.info(getClass(), "{} task aborted due to unlocking at {}/{}", key, index, total);
                }
                return locked;
            });
        }
        this.executorService.submit(() -> {
            try {
                task.run();
            } catch (Exception e) {
                LogUtil.error(getClass(), e);
            } finally {
                this.lockTemplate.release(lockKey);
            }
        });
    }

    public void submit(S key, boolean throwOnRunning, Runnable task) {
        submit(key, 0, throwOnRunning, task);
    }

    /**
     * 同步锁校验通过后同步运行指定任务
     *
     * @param key            同步锁键
     * @param throwOnRunning 当有正在运行的相同任务时是否抛出异常
     * @param task           任务
     */
    public void run(S key, long expiredMilliseconds, boolean throwOnRunning, Runnable task) {
        String lockKey = getLockKey(key);
        if (!this.lockTemplate.tryLock(lockKey, expiredMilliseconds)) {
            if (throwOnRunning) {
                Exception exception = getLockedException(key);
                if (exception != null) {
                    throw ExceptionUtil.toRuntimeException(exception);
                }
            } else {
                LogUtil.info(getClass(), "{}({}) is running, break", key, lockKey);
            }
            return;
        }
        try {
            task.run();
        } finally {
            this.lockTemplate.release(lockKey);
        }
    }

    public void run(S key, boolean throwOnRunning, Runnable task) {
        run(key, 0, throwOnRunning, task);
    }

    protected String getLockKey(S key) {
        return key.toString();
    }

    protected abstract Exception getLockedException(S key);

}
